package com.mlog.njyjy.ndop.msg.relay.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mlog.njyjy.ndop.msg.common.support.DateTimeUtil;
import com.mlog.njyjy.ndop.msg.relay.SpringTestBase;
import com.mlog.njyjy.ndop.msg.relay.consumer.KafkaConsumer;

import cn.hutool.core.io.FileUtil;

import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.concurrent.ListenableFuture;

import java.io.File;
import java.util.Random;
import java.util.UUID;

/**
 * @FunctionName: KafkaProducer
 * @Description: TODO
 * @author: lijian
 * @date: 2020/7/13 15:56
 */
public class KafkaProducer extends SpringTestBase {

    @Autowired
    private KafkaTemplate kafkaTemplate;
    
    @Autowired
    private KafkaConsumer csm;
    
    @Test
    public void test() {
    	File f = new File("C:\\Users\\HEX\\Documents\\WeChat Files\\angelsub\\FileStorage\\File\\2020-09", "productprocess.txt");
		String data = FileUtil.readUtf8String(f);
		JSONArray array = JSON.parseArray(data);
		for(int i=0;i<array.size();i++) {
			JSONObject obj = array.getJSONObject(i);
			if("QPF".equals(obj.getString("type")))
				csm.insert(obj);
		}
			
    }

    @Test
    public void send() {
        System.out.println("kafka 生成数据");

        String topic = "productpreview";
        String[] sources = new String[]{"MICAPS4", "MESIS", "GOPASS", "SWAN", "TYMIS", "MOAP"};
        sendTo(topic, sources[1], 2);
//        for (int i = 0; i < 2; i++) {
//
//            try {
//                Thread.sleep(120000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }
    }

    public void sendTo(String topic, String source, int index) {
        String message = UUID.randomUUID().toString();
        String st = DateTimeUtil.nowStr();
        String json =
                "{\n" +
                        "\"topic\":\"" + topic + "\",\n" +
                        "\"source\":\"" + source + "\",\n" +
                        "\"sourceIP\":\"192.168.50.69\",\n" +
                        "\"type\":\"QPF\",\n" +
                        "\"PID\":\"QPF000800060001" + String.format("%04d", new Random().nextInt(9999)) + "\",\n" +
                        "\"ID\":\"" + String.format("%05d", index) + "\",\n" +
                        "\"datetime\":\"" + st + "\",\n" +
                        "\"fileNames\":\"rain2020090910.001\",\n" +
                        "\"absoluteDataName\":\"/disk/data/rain2020090910.001\",\n" +
                        "\"fileSizes\":\"509\",\n" +
                        "\"result\":0,\n" +
                        "\"resultDesc\":\"数据上传成功\",\n" +
                        "\"reserved\":\"success\"\n" +
                        "}";
        ListenableFuture k_msg = kafkaTemplate.send(topic, json);
        k_msg.addCallback(o -> System.out.println("seng-消息成功" + message),
                throwable -> System.out.println("消息发送失败：" + message));

    }
}
